/**
 * This file is a part of qloudgen-sec. 
 * You can redistribute qloudgen-sec and/or modify it under the terms of the Lesser GNU General Public License version 3. 
 * qloudgen-sec is distributed WITHOUT ANY WARRANTY. 
 *
 * See the Lesser GNU General Public License for more details at http://www.gnu.org/licenses/. 
 **/
package org.redisson;

import java.net.URI;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.lambdaworks.redis.RedisConnection;
import com.lambdaworks.redis.RedisException;

import org.redisson.cluster.ClusterConnectionManager;

/**
 * <p>Title: Redis cluster command support</p>
 * <p>Description: Wrapper redisson cluster connection command implementation.</p>
 * <p>Copyright: qloudgen/sec 2014</p>
 * @author <a href="mailto:hyq.dd.sun@gmail.com">glue.3M</a>
 */
abstract class ClusterCommand<T> {
	/**
	 * Log output.
	 */
	private final static Logger logger = LoggerFactory.getLogger( ClusterCommand.class );
	/**
	 *
	 */
	private final ClusterConnectionManager connectionManager;
	/**
	 * Only keep current thread max recursived value.
	 */
	private int redirections = 5;
	
	/**
	 * Custom constructor.
	 *
	 * @param connectionManager
	 */
	public ClusterCommand(ClusterConnectionManager connectionManager) {
		this.connectionManager = connectionManager;
	}
	
	/**
	 * Execute.
	 *
	 * @param connection
	 * @return T
	 */
    public abstract T execute(RedisConnection connection);
	
	/**
	 * Cluster command.
	 *
	 * @param name
	 * @return T
	 */
	public T run(String name) {
		return runWithRetries( name , redirections , false );
	}
	
	/**
	 * Cluster command.
	 *
	 * @param num
	 * @return T
	 */
	public T run(int num) {
		return runWithRetries( Integer.valueOf( num ) , redirections , false );
	}

	/**
	 * Cluster command.
	 *
	 * @param key
	 * @param redirections
	 * @param asking
	 * @return T
	 */
	private T runWithRetries(Object key , int redirections , boolean asking) {
		//
		logger.trace( "Current redis cluster redirections is...{}" , redirections );
		if ( redirections <= 0 ) {
			throw new RuntimeException( "Current redirections reached max value..." );
		}
		//
		URI address = null;
		RedisConnection connection = null;
		try {
			/**
			 * Get current address.
			 */
			address = connectionManager.getAddress( key );
			/**
			 * Acquire connection from cluster node pools.
			 */
			connection = connectionManager.getConnection( key );
			/**
			 * Need ask.
			 */
			if ( asking ) {
				logger.debug( "Current redis cluster asked...{}" , connection.clusterAsk() );
				asking = false;
			}
			/**
			 * Execute command.
			 */
			return execute( connection );
		}
		catch (RedisException e) {
			//
			connection = null;
			redirections = redirections - 1;
			String message = e.getMessage();
			logger.error( "Current redis cluster changed...{}" , message );
			/**
			 *
			 */
			if ( isTimeout( message ) ) {
				logger.error( "Current redis cluster will refreshing pool by timeout..." );
				/**
				 * Force refresh slots & pools.
				 */
				connectionManager.refresh( key , address );
				/**
				 * Run current command again.
				 */
				return runWithRetries( key , redirections , asking );
			}
			else if ( isClosed( message ) ) {
				logger.error( "Current redis cluster will retrying pool..." );
				/**
				 * Run current command again.
				 */
				return runWithRetries( key , redirections , asking );
			}
			else if ( isDown( message ) ) {
				logger.error( "Current redis cluster will waiting pool..." );
				/**
				 *
				 */
				synchronized( this ) {
					try {
						wait( 6 * 1000L );
					}
					catch (Exception a) {
						logger.error( "{}" , a );
					}
				}
				/**
				 * Run current command again.
				 */
				return runWithRetries( key , redirections , asking );
			}
			/**
			 *
			 */
			else if ( isMoved( message ) || isAsked( message ) ) {
				logger.error( "Current redis cluster will refreshing pool by redirection..." );
				//
				if ( isAsked( message ) ) {
					asking = true;
				}
				/**
				 * Force refresh slots & pools.
				 */
				connectionManager.refresh( message );
				/**
				 * Run current command again.
				 */
				return runWithRetries( key , redirections , asking );
			}
			else {
				//
				throw e;
			}
		}
		finally {
			/**
			 * Release connection for cluster node pools.
			 */
			connectionManager.release( key , connection );
		}
	}

	/**
	 * Returns cluster moved.
	 *
	 * @param message
	 * @return boolean
	 */
	private boolean isMoved(String message) {
		try {
			return message.indexOf( "MOVED" ) > -1;
		}
		catch (Throwable t) {
			logger.error( "\n{}" , t );
		}
		return false;
	}

	/**
	 * Returns cluster asked.
	 *
	 * @param message
	 * @return boolean
	 */
	private boolean isAsked(String message) {
		try {
			return message.indexOf( "ASK" ) > -1;
		}
		catch (Throwable t) {
			logger.error( "\n{}" , t );
		}
		return false;
	}
	
	/**
	 * Returns cluster timeout.
	 *
	 * @param message
	 * @return boolean
	 */
	private boolean isTimeout(String message) {
		try {
			return message.indexOf( "Command timed out" ) > -1;
		}
		catch (Throwable t) {
			logger.error( "\n{}" , t );
		}
		return false;
	}
	
	/**
	 * Returns connection closed.
	 *
	 * @param message
	 * @return boolean
	 */
	private boolean isClosed(String message) {
		try {
			return message.indexOf( "Connection closed" ) > -1 || message.indexOf( "Connection is closed" ) > -1;
		}
		catch (Throwable t) {
			logger.error( "\n{}" , t );
		}
		return false;
	}
	
	/**
	 * Returns connection closed.
	 *
	 * @param message
	 * @return boolean
	 */
	private boolean isDown(String message) {
		try {
			return message.indexOf( "CLUSTERDOWN" ) > -1;
		}
		catch (Throwable t) {
			logger.error( "\n{}" , t );
		}
		return false;
	}
}